package com.atguigu.day09;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

//DDL
public class Example7 {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamTableEnvironment streamTableEnvironment = StreamTableEnvironment.create(env, settings);
        streamTableEnvironment.executeSql(
                "CREATE TABLE clicks (`user` String, `url` String) " +
                        "WITH (" +
                        "'connector' = 'filesystem'," +
                        "'path' = 'D:/WebSite/flink/flink/java_flink/src/main/resources/file.csv'," +
                        "'format' = 'csv'" +
                        ")"
        );
        streamTableEnvironment
                .executeSql("CREATE TABLE ResultTable (`user` STRING, `cnt` BIGINT) " +
                        "WITH ('connector' = 'print')");

        streamTableEnvironment
                .executeSql("INSERT INTO ResultTable SELECT user, COUNT(url) as cnt FROM clicks GROUP BY user");
    }
}
